[SPARK-57858][SQL] Emit BIN BY scaled DISTRIBUTE columns as produced attributes#56930
[SPARK-57858][SQL] Emit BIN BY scaled DISTRIBUTE columns as produced attributes#56930vranes wants to merge 2 commits into
Conversation
vranes
force-pushed
the
bin-by-distribute-produced-attrs
branch
2 times, most recently
from
e417f4b to
6be4341
Compare
vranes
force-pushed
the
bin-by-distribute-produced-attrs
branch
from
6be4341 to
7e4c5c2
Compare
cloud-fan
left a comment
cloud-fan
left a comment
There was a problem hiding this comment.
0 blocking, 1 non-blocking, 1 nit.
Clean attribute-identity change that restores the ExprId invariant, mirroring the established Generate/Expand produced-attribute pattern; both DeduplicateRelations phases and the references/missingInput wiring check out.
Correctness (1)
- ResolveBinBySuite.scala:205: no test for multiple DISTRIBUTE columns, though the description claims one — see inline
Nits: 1 minor item (see inline comments).
Verification
Traced the identity change: output re-points each DISTRIBUTE child slot to a fresh-id, same-name/type/nullability produced attribute (OUTPUT_SCHEMA agrees — the scaled column shadows its input, only exprId/qualifier/metadata differ). references = expressions -- producedAttributes leaves {rangeStart, rangeEnd, distributeColumns}, all in child.outputSet, so missingInput stays empty. Both dedup phases renew scaledDistributeColumns consistently with getExprIds keyed on producedAttributes, and distributeColumns (data-field inputs) follow the child via the standard rewriteAttrs renewal — so a self-join over a shared subtree resolves. No rows/values change (execution stubbed), so this is a pure plan-shape correction.
| assert(bi.distributeColumns.map(_.exprId) == Seq(value.exprId)) | ||
| } | ||
|
|
||
| test("resolved BinBy emits the DISTRIBUTE column as a produced attribute replacing the input") { |
There was a problem hiding this comment.
The PR description lists "multiple DISTRIBUTE columns are each replaced in place with distinct fresh ids" as a tested case, but this suite has no such test: this one resolves a single column and asserts scaledDistributeColumns.map(_.exprId) == Seq(outValue.exprId), and the only multi-element distribute list (Seq(value, value) at line 322) is the duplicate-rejection negative test.
Multi-column is the case distributeReplacements (an AttributeMap over a zip) and child.output.map(getOrElse) exist to serve — worth a test with two distinct DISTRIBUTE columns at different schema positions, asserting each is replaced in place with a distinct fresh id. Non-blocking, but it closes the gap and the stale description claim.
There was a problem hiding this comment.
Added a test which covers the distributeReplacements multi-column path. Thanks!
| } | ||
|
|
||
| override def output: Seq[Attribute] = child.output ++ appendedAttributes | ||
| assert(distributeColumns.length == scaledDistributeColumns.length, |
There was a problem hiding this comment.
Minor: the timeZoneId invariant right above (line 1826) throws SparkException.internalError, while this equal-length invariant uses assert. Both are reasonable, but two internal-invariant mechanisms side by side in the same constructor reads slightly inconsistently — consider matching the neighbor for uniformity.
There was a problem hiding this comment.
Changed the timeZoneId guard to assert too. Neither can actually fire unless a bug is introduced so assert is more appropriate. Thanks!
vranes
force-pushed
the
bin-by-distribute-produced-attrs
branch
from
049abc8 to
8b7cda5
Compare
…attributes ### What changes were proposed in this pull request? The `BIN BY` relation operator (SPARK-57133) proportionally rescales its `DISTRIBUTE UNIFORM` columns. The logical `BinBy` node carried those columns through `child.output` with the child's own `ExprId`, even though execution rewrites their values. This PR makes the rescaled `DISTRIBUTE` columns *produced* attributes with fresh `ExprId`s (same names, types, nullability, and positions), shadowing the inputs, mirroring `Generate.generatorOutput`: - `BinBy` gains a `scaledDistributeColumns` field; `output` swaps each `DISTRIBUTE` input slot for its scaled counterpart in place, and `producedAttributes` includes them. The input `distributeColumns` stay on the node as the executor's read inputs but leave `output`. - `BinBy.scaledDistributeAttributes` mints the fresh attributes (qualifier and metadata dropped, matching `expr AS value` computed-value semantics). - `ResolveBinBy` mints them; `DeduplicateRelations` renews them in both phases so self-joins over a shared `BinBy` subtree resolve. The two `BinBy` constructor invariants (`timeZoneId` set iff LTZ range, and one scaled attribute per `DISTRIBUTE` column) are both internal `assert`s: neither is reachable from `ResolveBinBy`, which derives `timeZoneId` from the same `rangeStart` type and mints `scaledDistributeColumns` by mapping over `distributeColumns`. ### Why are the changes needed? Catalyst relies on the invariant that the same `ExprId` everywhere implies the same value. No other operator edits a value under a retained child attribute (`Generate` / `Window` / `Expand` / `Aggregate` all mint fresh ids for changed columns). Carrying the rescaled `DISTRIBUTE` column under the child's `ExprId` violated that: any rule reasoning on `ExprId` (predicate pushdown, constraint propagation, common-subexpression elimination) could read the pre-scale value. It is harmless today only because no such rule lists `BinBy`, but that safety is incidental, not designed. Minting fresh identities restores the invariant and lets a filter or sort on a `DISTRIBUTE` column bind to the scaled output. ### Does this PR introduce _any_ user-facing change? No. `BIN BY` is gated off by default (SPARK-57440) and its physical execution is still stubbed, so the operator is not usable end-to-end yet; this is an internal analyzer / plan-shape change. The output schema (column names, types, positions) is unchanged. ### How was this patch tested? `ResolveBinBySuite`, including new cases that the rescaled `DISTRIBUTE` columns are produced attributes shadowing the input, that multiple `DISTRIBUTE` columns are each replaced in place with distinct fresh ids, and that qualifier/metadata are dropped on the produced column; plus the existing self-join deduplication regression. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Anthropic) Closes #56930 from vranes/bin-by-distribute-produced-attrs. Authored-by: Nikolina Vraneš <nikolina.vranes@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit 3d8bbe8) Signed-off-by: Wenchen Fan <wenchen@databricks.com>
2 participants
What changes were proposed in this pull request?
The
BIN BYrelation operator (SPARK-57133) proportionally rescales itsDISTRIBUTE UNIFORMcolumns. The logicalBinBynode carried those columns throughchild.outputwith the child's ownExprId, even though execution rewrites their values.This PR makes the rescaled
DISTRIBUTEcolumns produced attributes with freshExprIds (same names, types, nullability, and positions), shadowing the inputs, mirroringGenerate.generatorOutput:BinBygains ascaledDistributeColumnsfield;outputswaps eachDISTRIBUTEinput slot for its scaled counterpart in place, andproducedAttributesincludes them. The inputdistributeColumnsstay on the node as the executor's read inputs but leaveoutput.BinBy.scaledDistributeAttributesmints the fresh attributes (qualifier and metadata dropped, matchingexpr AS valuecomputed-value semantics).ResolveBinBymints them;DeduplicateRelationsrenews them in both phases so self-joins over a sharedBinBysubtree resolve.The two
BinByconstructor invariants (timeZoneIdset iff LTZ range, and one scaled attribute perDISTRIBUTEcolumn) are both internalasserts: neither is reachable fromResolveBinBy, which derivestimeZoneIdfrom the samerangeStarttype and mintsscaledDistributeColumnsby mapping overdistributeColumns.Why are the changes needed?
Catalyst relies on the invariant that the same
ExprIdeverywhere implies the same value. No other operator edits a value under a retained child attribute (Generate/Window/Expand/Aggregateall mint fresh ids for changed columns). Carrying the rescaledDISTRIBUTEcolumn under the child'sExprIdviolated that: any rule reasoning onExprId(predicate pushdown, constraint propagation, common-subexpression elimination) could read the pre-scale value. It is harmless today only because no such rule listsBinBy, but that safety is incidental, not designed. Minting fresh identities restores the invariant and lets a filter or sort on aDISTRIBUTEcolumn bind to the scaled output.Does this PR introduce any user-facing change?
No.
BIN BYis gated off by default (SPARK-57440) and its physical execution is still stubbed, so the operator is not usable end-to-end yet; this is an internal analyzer / plan-shape change. The output schema (column names, types, positions) is unchanged.How was this patch tested?
ResolveBinBySuite, including new cases that the rescaledDISTRIBUTEcolumns are produced attributes shadowing the input, that multipleDISTRIBUTEcolumns are each replaced in place with distinct fresh ids, and that qualifier/metadata are dropped on the produced column; plus the existing self-join deduplication regression.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)